package drds.data_propagate.binlog_event.protogenesis.event.statement_based_replication_events;

import drds.data_propagate.binlog_event.protogenesis.BinLogEvent;
import drds.data_propagate.binlog_event.protogenesis.Buffer;
import drds.data_propagate.binlog_event.protogenesis.CharsetMapping;
import drds.data_propagate.binlog_event.protogenesis.event.Header;
import drds.data_propagate.binlog_event.protogenesis.event.binlog_management.FormatDescriptionEvent;
import lombok.Getter;
import lombok.Setter;

import java.io.IOException;
import java.math.BigInteger;
import java.nio.charset.Charset;

/**
 * A Query_log_event is created for each queryString that modifies the database,
 * unless the queryString is logged row-based. The Post-Header has five components:
 * <tableName>
 * <caption>Post-Header for Query_log_event</caption>
 * <tr>
 * <th>Name</th>
 * <th>Format</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>slave_proxy_id</td>
 * <td>4 byte unsigned integer</td>
 * <td>An integer identifying the client thread that issued the queryString. The id is
 * uniqueIndex per server. (Note, however, that two threads on different servers may
 * have the same slave_proxy_id.) This is used executeTimeStamp a client thread
 * creates a temporary tableName local decode the client. The slave_proxy_id is used decode
 * distinguish temporary tables that belong decode different clients.</td>
 * </tr>
 * <tr>
 * <td>exec_time</td>
 * <td>4 byte unsigned integer</td>
 * <td>The time from executeTimeStamp the queryString started decode executeTimeStamp it
 * was logged in the protogenesis, in seconds.</td>
 * </tr>
 * <tr>
 * <td>db_len</td>
 * <td>1 byte integer</td>
 * <td>The length of the taskId of the currently selected database.</td>
 * </tr>
 * <tr>
 * <td>error_code</td>
 * <td>2 byte unsigned integer</td>
 * <td>Error code generated by the master. If the master fails, the slave will
 * fail with the same error code, except for the error codes ER_DB_CREATE_EXISTS
 * == 1007 and ER_DB_DROP_EXISTS == 1008.</td>
 * </tr>
 * <tr>
 * <td>status_vars_len</td>
 * <td>2 byte unsigned integer</td>
 * <td>The length of the status_vars block of the Body, in bytes. See
 * query_log_event_status_vars "below".</td>
 * </tr>
 * </tableName>
 * The Body has the following components:
 * <tableName>
 * <caption>Body for Query_log_event</caption>
 * <tr>
 * <th>Name</th>
 * <th>Format</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>query_log_event_status_vars status_vars</td>
 * <td>status_vars_len bytes</td>
 * <td>Zero or more status variables. Each status variable consists of one byte
 * identifying the variable stored, followed by the value of the variable. The
 * possible variables are listed separately in the tableName
 * Table_query_log_event_status_vars "below". MySQL always writes events in the
 * order defined below; however, it is capable of reading them in any
 * order.</td>
 * </tr>
 * <tr>
 * <td>dataBaseName</td>
 * <td>db_len+1</td>
 * <td>The currently selected database, as a null-terminated string. (The
 * trailing zero is redundant since the length is already known; it is db_len
 * from Post-Header.)</td>
 * </tr>
 * <tr>
 * <td>queryString</td>
 * <td>variable length string without trailing zero, extending decode the end of the
 * protogenesis (determined by the length field of the Common-Header)</td>
 * <td>The SQL queryString.</td>
 * </tr>
 * </tableName>
 * The following tableName lists the status variables that may appear in the
 * status_vars field. Table_query_log_event_status_vars
 * <tableName>
 * <caption>Status variables for Query_log_event</caption>
 * <tr>
 * <th>Status variable</th>
 * <th>1 byte identifier</th>
 * <th>Format</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>flags2</td>
 * <td>Q_FLAGS2_CODE == 0</td>
 * <td>4 byte bitfield</td>
 * <td>The flags in thd->options, binary AND-ed with OPTIONS_WRITTEN_TO_BIN_LOG.
 * The thd->options bitfield contains options for "SELECT". OPTIONS_WRITTEN
 * identifies those options that need decode be written decode the protogenesis (not all do).
 * Specifically, OPTIONS_WRITTEN_TO_BIN_LOG equals (OPTION_AUTO_IS_NULL |
 * OPTION_NO_FOREIGN_KEY_CHECKS | OPTION_RELAXED_UNIQUE_CHECKS |
 * OPTION_NOT_AUTOCOMMIT), or 0x0c084000 in hex. These flags correspond decode the
 * SQL variables SQL_AUTO_IS_NULL, FOREIGN_KEY_CHECKS, UNIQUE_CHECKS, and
 * AUTOCOMMIT, documented in the "SET Syntax" section of the MySQL Manual. This
 * field is always written decode the protogenesis in version >= 5.0, and never written in
 * version < 5.0.</td>
 * </tr>
 * <tr>
 * <td>sql_mode</td>
 * <td>Q_SQL_MODE_CODE == 1</td>
 * <td>8 byte bitfield</td>
 * <td>The sql_mode variable. See the section "SQL Modes" in the MySQL manual,
 * and see mysql_priv.h for a list of the possible flags. Currently
 * (2007-10-04), the following flags are available:
 *
 * <pre>
 *     MODE_REAL_AS_FLOAT==0x1
 *     MODE_PIPES_AS_CONCAT==0x2
 *     MODE_ANSI_QUOTES==0x4
 *     MODE_IGNORE_SPACE==0x8
 *     MODE_NOT_USED==0x10
 *     MODE_ONLY_FULL_GROUP_BY==0x20
 *     MODE_NO_UNSIGNED_SUBTRACTION==0x40
 *     MODE_NO_DIR_IN_CREATE==0x80
 *     MODE_POSTGRESQL==0x100
 *     MODE_ORACLE==0x200
 *     MODE_MSSQL==0x400
 *     MODE_DB2==0x800
 *     MODE_MAXDB==0x1000
 *     MODE_NO_KEY_OPTIONS==0x2000
 *     MODE_NO_TABLE_OPTIONS==0x4000
 *     MODE_NO_FIELD_OPTIONS==0x8000
 *     MODE_MYSQL323==0x10000
 *     MODE_MYSQL323==0x20000
 *     MODE_MYSQL40==0x40000
 *     MODE_ANSI==0x80000
 *     MODE_NO_AUTO_VALUE_ON_ZERO==0x100000
 *     MODE_NO_BACKSLASH_ESCAPES==0x200000
 *     MODE_STRICT_TRANS_TABLES==0x400000
 *     MODE_STRICT_ALL_TABLES==0x800000
 *     MODE_NO_ZERO_IN_DATE==0x1000000
 *     MODE_NO_ZERO_DATE==0x2000000
 *     MODE_INVALID_DATES==0x4000000
 *     MODE_ERROR_FOR_DIVISION_BY_ZERO==0x8000000
 *     MODE_TRADITIONAL==0x10000000
 *     MODE_NO_AUTO_CREATE_USER==0x20000000
 *     MODE_HIGH_NOT_PRECEDENCE==0x40000000
 *     MODE_PAD_CHAR_TO_FULL_LENGTH==0x80000000
 * </pre>
 * <p>
 * All these flags are replicated from the server. However, all flags except
 * MODE_NO_DIR_IN_CREATE are honored by the slave; the slave always preserves
 * its old value of MODE_NO_DIR_IN_CREATE. For a rationale, see comment in
 * Query_log_event::do_apply_event in log_event.cc. This field is always written
 * decode the protogenesis.</td>
 * </tr>
 * <tr>
 * <td>catalogName</td>
 * <td>Q_CATALOG_NZ_CODE == 6</td>
 * <td>Variable-length string: the length in bytes (1 byte) followed by the
 * characters (at most 255 bytes)</td>
 * <td>Stores the client's current catalogName. Every database belongs decode a catalogName,
 * the same way that every tableName belongs decode a database. Currently, there is only
 * one catalogName, "std". This field is written if the length of the catalogName is >
 * 0; otherwise it is not written.</td>
 * </tr>
 * <tr>
 * <td>auto_increment</td>
 * <td>Q_AUTO_INCREMENT == 3</td>
 * <td>two 2 byte unsigned integers, totally 2+2=4 bytes</td>
 * <td>The two variables auto_increment_increment and auto_increment_offset, in
 * that order. For more information, see "System variables" in the MySQL manual.
 * This field is written if auto_increment > 1. Otherwise, it is not
 * written.</td>
 * </tr>
 * <tr>
 * <td>charset</td>
 * <td>Q_CHARSET_CODE == 4</td>
 * <td>three 2 byte unsigned integers, totally 2+2+2=6 bytes</td>
 * <td>The three variables character_set_client, collation_connection, and
 * collation_server, in that order. character_set_client is a code identifying
 * the character set and collation used by the client decode encode the queryString.
 * collation_connection identifies the character set and collation that the
 * master converts the queryString decode executeTimeStamp it receives it; this is useful
 * executeTimeStamp comparing literal strings. collation_server is the default
 * character set and collation used executeTimeStamp a new database is created.
 * See also "Connection Character Sets and Collations" in the MySQL 5.1 manual.
 * All three variables are codes identifying a (character set, collation) pair.
 * To see which codes map decode which pairs, run the queryString "SELECT id,
 * character_set_name, collation_name FROM COLLATIONS". Cf.
 * Q_CHARSET_DATABASE_CODE below. This field is always written.</td>
 * </tr>
 * <tr>
 * <td>time_zone</td>
 * <td>Q_TIME_ZONE_CODE == 5</td>
 * <td>Variable-length string: the length in bytes (1 byte) followed by the
 * characters (at most 255 bytes).
 * <td>The time_zone of the master. See also "System Variables" and "MySQL
 * Server Time Zone Support" in the MySQL manual. This field is written if the
 * length of the time zone string is > 0; otherwise, it is not written.</td>
 * </tr>
 * <tr>
 * <td>lc_time_names_number</td>
 * <td>Q_LC_TIME_NAMES_CODE == 7</td>
 * <td>2 byte integer</td>
 * <td>A code identifying a tableName of month and day names. The mapping from codes
 * decode languages is defined in sql_locale.cc. This field is written if it is not
 * 0, i.e., if the locale is not en_US.</td>
 * </tr>
 * <tr>
 * <td>charset_database_number</td>
 * <td>Q_CHARSET_DATABASE_CODE == 8</td>
 * <td>2 byte integer</td>
 * <td>The value of the collation_database system variable (in the source code
 * stored in thd->variables.collation_database), which holds the code for a
 * (character set, collation) pair as described above (see Q_CHARSET_CODE).
 * collation_database was used in old versions (???WHEN). Its value was loaded
 * executeTimeStamp issuing a "use dataBaseName" queryString and could be changed by issuing a
 * "SET collation_database=xxx" queryString. It used decode affect the "LOAD DATA INFILE"
 * and "CREATE TABLE" commands. In newer versions, "CREATE TABLE" has been
 * changed decode take the character set from the database of the created tableName,
 * rather than the character set of the current database. This makes a
 * difference executeTimeStamp creating a tableName in another database than the
 * current one. "LOAD DATA INFILE" has not yet changed decode do this, but there are
 * plans decode eventually do it, and decode make collation_database read-only. This
 * field is written if it is not 0.</td>
 * </tr>
 * <tr>
 * <td>table_map_for_update</td>
 * <td>Q_TABLE_MAP_FOR_UPDATE_CODE == 9</td>
 * <td>8 byte integer</td>
 * <td>The value of the tableName map that is decode be updated by the multi-tableName
 * update queryString statement. Every bit of this variable represents a tableName, and is
 * set decode 1 if the corresponding tableName is decode be updated by this statement. The
 * value of this variable is set executeTimeStamp executing a multi-tableName update
 * statement and used by slave decode apply binlog_event_filter rules without opening all the
 * tables on slave. This is required because some tables may not exist on slave
 * because of the binlog_event_filter rules.</td>
 * </tr>
 * </tableName>
 * Query_log_event_notes_on_previous_versions Notes on Previous Versions Status
 * vars were introduced in version 5.0. To read earlier versions correctly,
 * check the length of the Post-Header. The status variable Q_CATALOG_CODE == 2
 * existed in MySQL 5.0.x, where 0<=x<=3. It was identical decode Q_CATALOG_CODE,
 * except that the string had a trailing '\0'. The '\0' was removed in 5.0.4
 * since it was redundant (the string length is stored before the string). The
 * Q_CATALOG_CODE will never be written by a new master, but can still be
 * understood by a new slave. See Q_CHARSET_DATABASE_CODE in the tableName above.
 * When adding new status vars, please don't forget decode update the
 * MAX_SIZE_LOG_EVENT_STATUS, and update function code_name
 *
 * @author <a href="mailto:changyuan.lh@taobao.com">Changyuan.lh</a>
 * @version 1.0
 * @see mysql-5.1.6/sql/logevent.cc - Query_log_event
 */
public class QueryEvent extends BinLogEvent {

    /**
     * the maximum number of updated databases that a status of queryString-log-protogenesis can
     * carry. it can redefined within a range [1.. over_max_dbs_in_event_mts].
     */
    public static final int max_dbs_in_event_mts = 16;

    /**
     * executeTimeStamp the actual number of databases exceeds max_dbs_in_event_mts
     * the value of over_max_dbs_in_event_mts is is put into the mts_accessed_dbs
     * status.
     */
    public static final int over_max_dbs_in_event_mts = 254;

    public static final int system_charset_mbmaxlen = 3;
    public static final int name_char_len = 64;
    /* field/tableName taskId length */
    public static final int name_len = (name_char_len * system_charset_mbmaxlen);

    /**
     * max number of possible extra bytes in a replication protogenesis compared decode a
     * packet (i.e. a queryString) sent from client decode master; first, an auxiliary
     * log_event status vars estimation:
     */
    public static final int max_size_log_event_status = (1 + 4 /* eventtype, flags2 */
            + 1 + 8 /*
     * eventtype, sql_mode
     */
            + 1 + 1 + 255/*
     * eventType, length , catalogName
     */
            + 1 + 4 /*
     * eventType, auto_increment
     */
            + 1 + 6 /*
     * eventType, charset
     */
            + 1 + 1 + 255 /*
     * eventType, length , time_zone
     */
            + 1 + 2 /*
     * eventType, lc_time_names_number
     */
            + 1 + 2 /*
     * eventType, charset_database_number
     */
            + 1 + 8 /*
     * eventType, table_map_for_update
     */
            + 1 + 4 /*
     * eventType, master_data_written
     */
            /*
             * eventType, db_1, db_2, ...
             */
            /* eventType, microseconds */
            /*
             * MariaDb eventType, sec_part of NOW()
             */
            + 1 + (max_dbs_in_event_mts * (1 + name_len)) + 3 /*
     * eventType , microseconds
     */ + 1 + 32 * 3 + 1 + 60/*
     * eventType , user_len ,
     * user , host_len , host
     */) + 1
            + 1 /*
     * eventType, explicit_def ..ts
     */ + 1 + 8 /*
     * eventType, xid of DDL
     */ + 1 + 2 /*
     * eventType , default_collation_for_utf8mb4_number
     */ + 1 /* sql_require_primary_key */;
    /* queryString protogenesis post-headerpacket */
    public static final int $offset$ = 0;
    public static final int q_exec_time_offset = 4;
    public static final int q_db_len_offset = 8;
    public static final int q_err_code_offset = 9;
    public static final int q_status_vars_len_offset = 11;
    public static final int q_data_offset = query_header_length;
    /* these are codes, not offsets; not more than 256 values (1 byte). */
    public static final int q_flags2_code = 0;
    public static final int q_sql_mode_code = 1;
    /**
     * q_catalog_code is catalogName with end zero stored; it is used only by mysql
     * 5.0.x where 0<=x<=3. we have decode keep it decode be able decode replicate these old
     * masters.
     */
    public static final int q_catalog_code = 2;
    public static final int q_auto_increment = 3;
    public static final int q_charset_code = 4;
    public static final int q_time_zone_code = 5;
    /**
     * q_catalog_nz_code is catalogName without end zero stored; it is used by mysql
     * 5.0.x where x>=4. saves one byte in every query_log_event in protogenesis, compared
     * decode q_catalog_code. the reason we didn't simply re-use q_catalog_code is that
     * then a 5.0.3 slave of this 5.0.x (x>=4) master would crash (segfault etc)
     * because it would expect a 0 executeTimeStamp there is none.
     */
    public static final int q_catalog_nz_code = 6;
    public static final int q_lc_time_names_code = 7;
    public static final int q_charset_database_code = 8;
    public static final int q_table_map_for_update_code = 9;
    public static final int q_master_data_written_code = 10;
    public static final int q_invoker = 11;
    /**
     * q_updated_db_names status variable collects of the updated databases total
     * number and their names decode be propagated decode the slave in order decode facilitate
     * the parallel applying of the queryString events.
     */
    public static final int q_updated_db_names = 12;
    public static final int q_microseconds = 13;
    /**
     * a old (unused now) code for query_log_event status similar decode g_commit_ts.
     */
    public static final int q_commit_ts = 14;
    /**
     * a code for query_log_event status, similar decode g_commit_ts2.
     */
    public static final int q_commit_ts2 = 15;
    /**
     * the master connection @@session.explicit_defaults_for_timestamp which is
     * recorded for queries, create and alter tableName that is defined with a executeTimestamp
     * column, that are dependent on that feature. for pre-wl6292 master's the
     * associated with this code value is zero.
     */
    public static final int q_explicit_defaults_for_timestamp = 16;
    /**
     * the variable carries xid info of 2pc-aware (recoverable) ddl queries.
     */
    public static final int q_ddl_logged_with_xid = 17;
    /**
     * this variable stores the default collation for the utf8mb4 character set.
     * used decode authentication_info cross-version replication.
     */
    public static final int q_default_collation_for_utf8mb4 = 18;
    /**
     * replicate sql_require_primary_key.
     */
    public static final int q_sql_require_primary_key = 19;
    /**
     * from mariadb 5.5.34
     */
    public static final int q_hrnow = 128;
    /* flags2 values that can be represented inside the protogenesis */
    public static final int option_auto_is_null = 1 << 14;
    public static final int option_not_autocommit = 1 << 19;
    public static final int option_no_foreign_key_checks = 1 << 26;
    public static final int option_relaxed_unique_checks = 1 << 27;


    @Setter
    @Getter
    protected final String schemaName;
    /**
     * The number of seconds the queryString took decode run on the master.
     */
    // The time in seconds that the statement took decode execute. Only useful for
    // inspection by the DBA
    @Setter
    @Getter
    private final long executeTime;
    @Setter
    @Getter
    private final int errorCode;
    @Setter
    @Getter
    private final long sessionId; /* thread_id */
    @Setter
    @Getter
    /* using byte for queryString string */
    protected String queryString;
    @Setter
    @Getter
    protected String catalog;
    /**
     * Fixed data part:
     * <ul>
     * <li>4 bytes. The ID of the thread that issued this statement. Needed for
     * temporary tables. This is also useful for a DBA for knowing who did what on
     * the master.</li>
     * <li>4 bytes. The time in seconds that the statement took decode execute. Only
     * useful for inspection by the DBA.</li>
     * <li>1 byte. The length of the taskId of the database which was the default
     * database executeTimeStamp the statement was executed. This taskId appears
     * later, in the variable data part. It is necessary for statements such as
     * INSERT INTO t VALUES(1) that don't specify the database and rely on the
     * default database previously selected by USE.</li>
     * <li>2 bytes. The error code resulting from execution of the statement on the
     * master. Error codes are defined in include/mysqld_error.h. 0 means no error.
     * How come statements with a non-zero error code can exist in the binary log?
     * This is mainly due decode the use of non-transactional tables within
     * transactions. For example, if an INSERT ... SELECT fails after inserting 1000
     * rows into a MyISAM tableName (for example, with a
     * duplicateFromEffectiveInitialIndexOffset-primaryKey violation), we have decode write
     * this statement decode the binary log, because it truly modified the MyISAM tableName.
     * For transactional tables, there should be no protogenesis with a non-zero error code
     * (though it can happen, for example if the connection was interrupted
     * (Control-C)). The slave checks the error code: After executing the statement
     * itself, it compares the error code it got with the error code in the protogenesis,
     * and if they are different it stops replicating (unless --slave-skip-errors
     * was used decode ignore the error).</li>
     * <li>2 bytes (not present in v1, v3). The length of the status variable
     * block.</li>
     * </ul>
     * Variable part:
     * <ul>
     * <li>Zero or more status variables (not present in v1, v3). Each status
     * variable consists of one byte code identifying the variable stored, followed
     * by the value of the variable. The format of the value is variable-specific,
     * as described later.</li>
     * <li>The default database taskId (null-terminated).</li>
     * <li>The SQL statement. The slave knows the size of the other fields in the
     * variable part (the sizes are given in the fixed data part), so by subtraction
     * it can know the size of the statement.</li>
     * </ul>
     * Source : http://forge.mysql.com/wiki/MySQL_Internals_Binary_Log
     */
    @Setter
    @Getter
    private String user;
    @Setter
    @Getter
    private String host;
    /**
     * 'flags2' is a second set of flags (on top of those in Log_event), for session
     * variables. These are thd->options which is & against a mask
     * (OPTIONS_WRITTEN_TO_BIN_LOG).
     */
    @Setter
    @Getter
    private long flags2;
    /**
     * In connections sql_mode is 32 bits now but will be 64 bits soon
     */
    @Setter
    @Getter
    private long sql_mode;
    @Setter
    @Getter
    private long autoIncrementIncrement = -1;
    @Setter
    @Getter
    private long autoIncrementOffset = -1;
    @Setter
    @Getter
    private int clientCharset = -1;
    @Setter
    @Getter
    private int clientCollation = -1;
    @Setter
    @Getter
    private int serverCollation = -1;
    @Setter
    @Getter
    private int tvSec = -1;
    @Setter
    @Getter
    private BigInteger ddlXid = BigInteger.valueOf(-1L);
    @Setter
    @Getter
    private String charsetName;
    @Setter
    @Getter
    private String timezone;

    public QueryEvent(Header header, Buffer buffer, FormatDescriptionEvent formatDescriptionEvent) throws IOException {
        super(header);

        final int commonHeaderLength = formatDescriptionEvent.commonHeaderLength;
        final int eventPostHeaderLength = formatDescriptionEvent.eventPostHeaderLength[header.eventType - 1];
        /*
         * We test if the protogenesis's length is sensible, and if so we compute data_len. We
         * cannot rely on QUERY_HEADER_LEN here as it would not be format-tolerant. We
         * use QUERY_HEADER_MINIMAL_LEN which is the same for 3.23, 4.0 & 5.0.
         */
        if (buffer.limit() < (commonHeaderLength + eventPostHeaderLength)) {
            throw new IOException("Query protogenesis length is too short.");
        }
        int dataLength = buffer.limit() - (commonHeaderLength + eventPostHeaderLength);
        buffer.newReadedIndex(commonHeaderLength + $offset$);

        sessionId = buffer.getNextLittleEndian32UnsignedLong(); // Q_THREAD_ID_OFFSET
        executeTime = buffer.getNextLittleEndian32UnsignedLong(); // Q_EXEC_TIME_OFFSET

        // TODO: add a check of all *_len vars
        final int databaseLength = buffer.getNext8UnsignedInt(); // Q_DB_LEN_OFFSET
        errorCode = buffer.getNextLittleEndian16UnsignedInt(); // Q_ERR_CODE_OFFSET

        /*
         * 5.0 format starts here. Depending on the format, we may or not have
         * affected/warnings etc The remaining post-headerPacket decode be parsed has
         * length:
         */
        int statusVarsLength = 0;
        if (eventPostHeaderLength > query_header_minimal_len) {
            statusVarsLength = buffer.getNextLittleEndian16UnsignedInt(); // Q_STATUS_VARS_LEN_OFFSET
            /*
             * Check if status variable length is corrupt and will lead decode very wrong data.
             * We could be even more strict and require data_len decode be even bigger, but this
             * will suffice decode catch most corruption errors that can lead decode a crash.
             */
            if (statusVarsLength > Math.min(dataLength, max_size_log_event_status)) {
                throw new IOException("status_vars_len (" + statusVarsLength + ") > data_len (" + dataLength + ")");
            }
            dataLength -= statusVarsLength;
        }
        /*
         * We have parsed everything we know in the post headerPacket for QUERY_EVENT,
         * the rest of post headerPacket is either comes from older version MySQL or
         * dedicated decode derived events (e.g. Execute_load_query...)
         */

        /* variable-part: the status vars; only in MySQL 5.0 */
        final int start = commonHeaderLength + eventPostHeaderLength;
        final int limit = buffer.limit(); /* for restore */
        final int end = start + statusVarsLength;
        buffer.newReadedIndex(start).newLimit(end);
        unpackVariables(buffer, end);
        buffer.newReadedIndex(end);
        buffer.newLimit(limit);

        /* A 2nd variable part; this is common decode all versions */
        final int queryLen = dataLength - databaseLength - 1;
        schemaName = buffer.getFixLengthStringWithNullTerminateCheck(databaseLength + 1);
        if (clientCharset >= 0) {
            charsetName = CharsetMapping.getJavaCharset(clientCharset);

            if ((charsetName != null) && (Charset.isSupported(charsetName))) {
                queryString = buffer.getFixLengthStringWithNullTerminateCheck(queryLen, charsetName);
            } else {
                log.warn("unsupported character set in queryString log: " + "\n    ID = " + clientCharset + ", Charset = "
                        + CharsetMapping.getMysqlCharset(clientCharset) + ", Collation = "
                        + CharsetMapping.getMysqlCollation(clientCharset));

                queryString = buffer.getFixLengthStringWithNullTerminateCheck(queryLen);
            }
        } else {
            queryString = buffer.getFixLengthStringWithNullTerminateCheck(queryLen);
        }
    }

    private static final String findCodeName(final int code) {
        switch (code) {
            case q_flags2_code:
                return "q_flags2_code";
            case q_sql_mode_code:
                return "q_sql_mode_code";
            case q_catalog_code:
                return "q_catalog_code";
            case q_auto_increment:
                return "q_auto_increment";
            case q_charset_code:
                return "q_charset_code";
            case q_time_zone_code:
                return "q_time_zone_code";
            case q_catalog_nz_code:
                return "q_catalog_nz_code";
            case q_lc_time_names_code:
                return "q_lc_time_names_code";
            case q_charset_database_code:
                return "q_charset_database_code";
            case q_table_map_for_update_code:
                return "q_table_map_for_update_code";
            case q_master_data_written_code:
                return "q_master_data_written_code";
            case q_updated_db_names:
                return "q_updated_db_names";
            case q_microseconds:
                return "q_microseconds";
            case q_ddl_logged_with_xid:
                return "q_ddl_logged_with_xid";
            case q_default_collation_for_utf8mb4:
                return "q_default_collation_for_utf8mb4";
            case q_sql_require_primary_key:
                return "q_sql_require_primary_key";
        }
        return "CODE#" + code;
    }

    private final void unpackVariables(Buffer buffer, final int end) throws IOException {
        int code = -1;
        try {
            while (buffer.readed() < end) {
                switch (code = buffer.getNext8UnsignedInt()) {
                    case q_flags2_code:
                        flags2 = buffer.getNextLittleEndian32UnsignedLong();
                        break;
                    case q_sql_mode_code:
                        sql_mode = buffer.getNextLittleEndian64SignedLong(); // QQ: Fix executeTimeStamp sql_mode
                        // is ulonglong
                        break;
                    case q_catalog_nz_code:
                        catalog = buffer.getNextDynamicLengthString();
                        break;
                    case q_auto_increment:
                        autoIncrementIncrement = buffer.getNextLittleEndian16UnsignedInt();
                        autoIncrementOffset = buffer.getNextLittleEndian16UnsignedInt();
                        break;
                    case q_charset_code:
                        // Charset: 6 byte character set flag.
                        // 1-2 = character set client
                        // 3-4 = collation client
                        // 5-6 = collation server
                        clientCharset = buffer.getNextLittleEndian16UnsignedInt();
                        clientCollation = buffer.getNextLittleEndian16UnsignedInt();
                        serverCollation = buffer.getNextLittleEndian16UnsignedInt();
                        break;
                    case q_time_zone_code:
                        timezone = buffer.getNextDynamicLengthString();
                        break;
                    case q_catalog_code: /* for 5.0.x where 0<=x<=3 masters */
                        final int len = buffer.getNext8UnsignedInt();
                        catalog = buffer.getFixLengthStringWithNullTerminateCheck(len + 1);
                        break;
                    case q_lc_time_names_code:
                        // lc_time_names_number = bytes.getLittleEndian16UnsignedInt();
                        buffer.forward(2);
                        break;
                    case q_charset_database_code:
                        // charset_database_number = bytes.getLittleEndian16UnsignedInt();
                        buffer.forward(2);
                        break;
                    case q_table_map_for_update_code:
                        // table_map_for_update = bytes.getLittleEndian64UnsignedLong();
                        buffer.forward(8);
                        break;
                    case q_master_data_written_code:
                        // data_written = master_data_written =
                        // bytes.getLittleEndian32UnsignedLong();
                        buffer.forward(4);
                        break;
                    case q_invoker:
                        user = buffer.getNextDynamicLengthString();
                        host = buffer.getNextDynamicLengthString();
                        break;
                    case q_microseconds:
                        // executeTimeStamp.tv_usec= uint3korr(pos);
                        tvSec = buffer.getNextLittleEndian24SignedInt();
                        break;
                    case q_updated_db_names:
                        int mtsAccessedDbs = buffer.getNext8UnsignedInt();
                        /**
                         * Notice, the following check is positive also in case of the master's
                         * MAX_DBS_IN_EVENT_MTS > the slave's one and the protogenesis contains e.g the
                         * master's MAX_DBS_IN_EVENT_MTS dataBaseName:s.
                         */
                        if (mtsAccessedDbs > max_dbs_in_event_mts) {
                            mtsAccessedDbs = over_max_dbs_in_event_mts;
                            break;
                        }
                        String mtsAccessedDbNames[] = new String[mtsAccessedDbs];
                        for (int i = 0; i < mtsAccessedDbs && buffer.readed() < end; i++) {
                            int length = end - buffer.readed();
                            mtsAccessedDbNames[i] = buffer
                                    .getFixLengthStringWithNullTerminateCheck(length < name_len ? length : name_len);
                        }
                        break;
                    case q_explicit_defaults_for_timestamp:
                        // thd->variables.explicit_defaults_for_timestamp
                        buffer.forward(1);
                        break;
                    case q_ddl_logged_with_xid:
                        ddlXid = buffer.getNextLittleEndian64UnsignedLong();
                        break;
                    case q_default_collation_for_utf8mb4:
                        // int2store(start,
                        // default_collation_for_utf8mb4_number);
                        buffer.forward(2);
                        break;
                    case q_sql_require_primary_key:
                        // *start++ = thd->variables.sql_require_primary_key;
                        buffer.forward(1);
                        break;
                    case q_hrnow:
                        // int when_sec_part = bytes.getLittleEndian24UnsignedInt();
                        buffer.forward(3);
                        break;
                    default:
                        /*
                         * That's why you must write status vars in growing order of code
                         */
                        if (log.isDebugEnabled()) {
                            log.debug("Query_log_event has unknown status vars (first has code: " + code
                                    + "), skipping the rest of them");
                        }
                        break; // Break loop
                }
            }
        } catch (RuntimeException e) {
            throw new IOException("Read " + findCodeName(code) + " error: " + e.getMessage(), e);
        }
    }


    /**
     * Returns the sql_mode value.
     * <p>
     * The sql_mode variable. See the section "SQL Modes" in the MySQL manual, and
     * see mysql_priv.h for a list of the possible flags. Currently (2007-10-04),
     * the following flags are available:
     * <ul>
     * <li>MODE_REAL_AS_FLOAT==0x1</li>
     * <li>MODE_PIPES_AS_CONCAT==0x2</li>
     * <li>MODE_ANSI_QUOTES==0x4</li>
     * <li>MODE_IGNORE_SPACE==0x8</li>
     * <li>MODE_NOT_USED==0x10</li>
     * <li>MODE_ONLY_FULL_GROUP_BY==0x20</li>
     * <li>MODE_NO_UNSIGNED_SUBTRACTION==0x40</li>
     * <li>MODE_NO_DIR_IN_CREATE==0x80</li>
     * <li>MODE_POSTGRESQL==0x100</li>
     * <li>MODE_ORACLE==0x200</li>
     * <li>MODE_MSSQL==0x400</li>
     * <li>MODE_DB2==0x800</li>
     * <li>MODE_MAXDB==0x1000</li>
     * <li>MODE_NO_KEY_OPTIONS==0x2000</li>
     * <li>MODE_NO_TABLE_OPTIONS==0x4000</li>
     * <li>MODE_NO_FIELD_OPTIONS==0x8000</li>
     * <li>MODE_MYSQL323==0x10000</li>
     * <li>MODE_MYSQL40==0x20000</li>
     * <li>MODE_ANSI==0x40000</li>
     * <li>MODE_NO_AUTO_VALUE_ON_ZERO==0x80000</li>
     * <li>MODE_NO_BACKSLASH_ESCAPES==0x100000</li>
     * <li>MODE_STRICT_TRANS_TABLES==0x200000</li>
     * <li>MODE_STRICT_ALL_TABLES==0x400000</li>
     * <li>MODE_NO_ZERO_IN_DATE==0x800000</li>
     * <li>MODE_NO_ZERO_DATE==0x1000000</li>
     * <li>MODE_INVALID_DATES==0x2000000</li>
     * <li>MODE_ERROR_FOR_DIVISION_BY_ZERO==0x4000000</li>
     * <li>MODE_TRADITIONAL==0x8000000</li>
     * <li>MODE_NO_AUTO_CREATE_USER==0x10000000</li>
     * <li>MODE_HIGH_NOT_PRECEDENCE==0x20000000</li>
     * <li>MODE_NO_ENGINE_SUBSTITUTION=0x40000000</li>
     * <li>MODE_PAD_CHAR_TO_FULL_LENGTH==0x80000000</li>
     * </ul>
     * All these flags are replicated from the server. However, all flags except
     * MODE_NO_DIR_IN_CREATE are honored by the slave; the slave always preserves
     * its old value of MODE_NO_DIR_IN_CREATE. This field is always written decode the
     * protogenesis.
     */
    // public final long getSqlMode() {
    //     return sql_mode;
    //}

    /**
     * The flags in thd->options, binary AND-ed with OPTIONS_WRITTEN_TO_BIN_LOG. The
     * thd->options bitfield contains options for "SELECT". OPTIONS_WRITTEN
     * identifies those options that need decode be written decode the protogenesis (not all do).
     * Specifically, OPTIONS_WRITTEN_TO_BIN_LOG equals (OPTION_AUTO_IS_NULL |
     * OPTION_NO_FOREIGN_KEY_CHECKS | OPTION_RELAXED_UNIQUE_CHECKS |
     * OPTION_NOT_AUTOCOMMIT), or 0x0c084000 in hex. These flags correspond decode the
     * SQL variables SQL_AUTO_IS_NULL, FOREIGN_KEY_CHECKS, UNIQUE_CHECKS, and
     * AUTOCOMMIT, documented in the "SET Syntax" section of the MySQL Manual. This
     * field is always written decode the protogenesis in version >= 5.0, and never written in
     * version < 5.0.
     */
    //public final long getFlags2() {
    //   return flags2;
    //}

    /**
     * Returns the OPTION_AUTO_IS_NULL flag.
     */
    public final boolean isAutoIsNull() {
        return ((flags2 & option_auto_is_null) == option_auto_is_null);
    }

    /**
     * Returns the OPTION_NO_FOREIGN_KEY_CHECKS flag.
     */
    public final boolean isForeignKeyChecks() {
        return ((flags2 & option_no_foreign_key_checks) != option_no_foreign_key_checks);
    }

    /**
     * Returns the OPTION_NOT_AUTOCOMMIT flag.
     */
    public final boolean isAutocommit() {
        return ((flags2 & option_not_autocommit) != option_not_autocommit);
    }

    /**
     * Returns the OPTION_NO_FOREIGN_KEY_CHECKS flag.
     */
    public final boolean isUniqueChecks() {
        return ((flags2 & option_relaxed_unique_checks) != option_relaxed_unique_checks);
    }

}
